/*
* @Description: 上报进程
* @Author: Sunqk5665
* @Date: 2023-11-6
* @LastEditTime: 2023-11-13
*/
#include <stdio.h>
#include <unistd.h>
#include <stdlib.h>
#include "cJSON.h"
#include <string.h>
#include <pthread.h>
#include "MQTTClient.h"
#include "shmem.h"          // 共享内存头文件
#include "msg_queue_peer.h" // 消息队列头文件
#include <sqlite3.h>
#include <time.h>

// #define ADDRESS "tcp://192.168.51.58:1883"
#define CLIENTID "ExampleClientSub"
#define TOPIC1 "/app/data/down"
#define TOPIC2 "/app/data/up" // 数据上报主题
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L

void *upHandler(void *arg); // 数据上报线程函数
void flashUp(); // 刷新上报函数
void changeUp(); // 数据改变上报函数
void circleUp(); // 循环上报函数
void *storeHandler(void *arg);    // 数据存储线程函数
void *instructHandler(void *arg); // 指令处理线程函数

void historysearch(); // 历史数据查询函数
void modeChange(); // 模式修改函数
void ctrlFunc(); // 控制指令函数

static struct shm_param para;
union val_t {
    int b_val;   //bool类型存储空间
    int i_val;   //整形值存储空间
    float f_val; //浮点值存储空间
};
struct std_node
{
    int key;             //唯一键值
    int type;            //数据点类型（0:bool 1:int 2:float）
    int dev_type;        //数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val; //变化上报后需要更新旧值
    union val_t new_val; //从共享内存取出最新数据，放到new_val中
    int ret;             //默认为-1，采集成功后设置为0，采集失败再置-1
};
static int *total;                //指向共享内存中数据节点总个数
static struct std_node *node_arr; //指向共享内存中节点缓存数组头

struct msgbuf
{
    long mtype; // 消息队列类型
    char mdata[256];
};
struct msgbuf send_buf; 
#define MAX_NODE 10 //最大支持设备额数目
#define STD_NODE_LEN sizeof(struct std_node)
#define MAX_NODE_SIZE (MAX_NODE * STD_NODE_LEN)
/**全局变量：指令类型标志**/
#define CAIJI 1
#define KONGZHI 2
#define MOSHI 3
#define LISHI 4
int insFlag = 0;
/**********************/

int ctrlkey;
int changetype, changeperiod;
int historykey; // 历史查询中的设备key
long int startTime, endTime;

volatile MQTTClient_deliveryToken deliveredtoken;

// 客户端句柄（描述符）
MQTTClient client;

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

/**
 * @brief 
 * 
 * @param context 
 * @param topicName 收到来自哪个主题的消息
 * @param topicLen 主题的长度
 * @param message 消息体:payload(消息体，字符串)  payloadlen：消息的长度
 * @return int 
 */
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: \n");
    // 客户端指令解析（反序列化）
    cJSON *root = cJSON_Parse((char *)message->payload); // 把字符串转换成json树
    if (NULL == root)
    {
        printf("parse err\n");
        return -1;
    }
    cJSON *type = cJSON_GetObjectItem(root, "type");
    if (type->valueint == 1) // 采集指令/采集回复指令
    {
        printf("=====采集指令/采集回复指令=====\n");
        printf("%s\n",(char *)message->payload); // 打印一下Qt发来的json字符串
        
        insFlag = 1;
    }
    else if (type->valueint == 2) // 控制指令/控制回复指令
    {
        printf("=====控制指令/控制回复指令=====\n");
        // 解析携带的数据
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        cJSON *val = cJSON_GetObjectItem(data, "val");
        // struct msgbuf send_buf;
        strcpy(send_buf.mdata, (char *)message->payload);
        printf("%s\n", send_buf.mdata);
        // 拿到设备的key值，赋值给全局变量
        ctrlkey = key->valueint;

        insFlag = 2;
    }
    else if (type->valueint == 3) // 模式修改
    {
        printf("========模式修改指令========\n");
        printf("%s\n",(char *)message->payload);
        //客户需要修改当前的上报方式，需要解析指令并将新的模式写到配置文件中，等待下次重启生效
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *type = cJSON_GetObjectItem(data, "type");
        cJSON *period = cJSON_GetObjectItem(data, "period");
        changetype = type->valueint;
        changeperiod = period->valueint;

        insFlag = 3;
    }
    else if (type->valueint == 4) // 历史数据查询
    {
        printf("=======历史数据查询指令=======\n");
        printf("%s\n",(char *)message->payload);
        cJSON *data = cJSON_GetObjectItem(root, "data");
        cJSON *key = cJSON_GetObjectItem(data, "key");
        historykey = key->valueint;
        cJSON *limit = cJSON_GetObjectItem(data, "limit");
        // int	size = cJSON_GetArraySize(data); // 获取数组大小
        startTime = cJSON_GetArrayItem(limit, 0)->valueint; // 起始时间
        endTime = cJSON_GetArrayItem(limit, 1)->valueint;   // 结束时间

        insFlag = 4;
    }
    cJSON_Delete(root);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int type, period; // 点表中的上报类型以及周期上报的周期

int main(int argc, char *argv[])
{
    /***** 点表解析--json的反序列化 *****/
    // 读取点表文件的内容
    //FILE *file = fopen("node.json", "r");
    FILE *file;
    printf("检测点表文件中."); fflush(NULL);
    sleep(1);
    printf("."); fflush(NULL);
    sleep(1);
    printf(".\n");
    while (1)
    {
        if (access("node.json",F_OK)!=0) // 检测点表文件是否存在
        {
            perror("====点表文件不存在！====\n");
            sleep(2);
            //return -1;
        }else
        {
            file = fopen("node.json", "r");
            printf("====点表已发现！打开成功=====\n");
            sleep(2);
            break;
        }
    }
    char buf[2048] = {0};
    // size_t fread(void *ptr, size_t size, size_t nmemb, FILE *stream);
    size_t len = fread(buf, sizeof(char), 2048, file);
    if (len < 0)
    {
        perror("read err");
        return -1;
    }
    // printf("buf = %s\n", buf); // 打印测试
    cJSON *root = cJSON_Parse(buf);
    if (NULL == root)
    {
        printf("parse err\n");
        return -1;
    }
    // 获取树干上名字叫version的子树杈
    cJSON *item = cJSON_GetObjectItem(root, "version");
    printf("%s = %s\n", item->string, item->valuestring);
    printf("==========================\n");
    // 获取设备上报属性
    cJSON *report = cJSON_GetObjectItem(root, "report");
    item = cJSON_GetObjectItem(report, "type");
    type = item->valueint; // 上报类型
    printf("%s = %d\n", item->string, type);
    item = cJSON_GetObjectItem(report, "period");
    period = item->valueint; //上报周期时间，单位秒，仅在type=2时有效
    printf("%s = %d\n", item->string, period);
    printf("==========================\n");
    // 获取MQTT属性
    char mqttserverURI[30] = {0};
    sprintf(mqttserverURI, "%s", "tcp://");
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    printf("%s\n", mqtt_server->string);
    item = cJSON_GetObjectItem(mqtt_server, "addr");
    printf("%s = %s\n", item->string, item->valuestring);
    strcat(mqttserverURI, item->valuestring);
    strcat(mqttserverURI, ":");
    item = cJSON_GetObjectItem(mqtt_server, "port");
    printf("%s = %d\n", item->string, item->valueint);
    char port[6] = {0};
    sprintf(port, "%d", item->valueint);
    strcat(mqttserverURI, port);
    printf("%s\n", mqttserverURI);

    printf("==========================\n");
    fclose(file);

    /***** mqtt服务器start********/
    // // 客户端句柄（描述符）
    // MQTTClient client;
    // 连接参数
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    // 发布 配置
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;
    int ch;

    //创建客户端，并且指定客户端连接的mqtt服务器地址和客户端ID
    MQTTClient_create(&client, mqttserverURI, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    //初始化连接参数
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    //设置回调接口，只需要关注msgarrvd：消息到达后，会自动调用这个接口
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    //连接到broker
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("connect to MQTT server success!\n");
    //订阅某个主题，指定订阅主题的名字，可以指定qos服务质量
    MQTTClient_subscribe(client, TOPIC1, QOS);
    char msg[256] = {0};

    /************创建共享内存、解析点表*******/
    int ret = -1;
    ret = shm_init(&para, "shm_test", MAX_NODE_SIZE);
    if (ret < 0)
    {
        return -1;
    }
    void *node_p = shm_getaddr(&para); // 获取共享内存首地址
    if (node_p == NULL)
    {
        return -1;
    }
    memset(node_p, 0, MAX_NODE_SIZE);
    //前4个字节存储实际的设备数目
    total = (int *)node_p;
    //后面空间存储数据点
    node_arr = (struct std_node *)(node_p + sizeof(int));
    // *total = 6;  //假设有6个设备

    // stm32配置解析，将设备节点放入共享内存
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    printf("%s\n", stm32->string);
    cJSON *data = cJSON_GetObjectItem(stm32, "data");
    printf("%s\n", data->string);
    int size = cJSON_GetArraySize(data);
    for (int i = 0; i < size; i++) // 遍历取数组项
    {
        //取出每数组项中的节点，并解析
        cJSON *tmp = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(tmp, "key");
        cJSON *name = cJSON_GetObjectItem(tmp, "name");
        cJSON *type = cJSON_GetObjectItem(tmp, "type");
        node_arr[i].key = key->valueint; //唯一键值
        node_arr[i].type = type->valueint; //数据点类型
        node_arr[i].dev_type = 0; // 0:stm32（设备类型标志）
        // node_arr[i].old_val.f_val = 22.0;
        // node_arr[i].new_val.f_val = 15.0;
        node_arr[i].ret = -1; //默认为-1，采集成功后设置为0，采集失败再置-1
    }
    int index = size;
    // modbus配置解析,将设备节点放入共享内存
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    printf("%s\n", modbus->string);
    cJSON *name = cJSON_GetObjectItem(modbus, "name");
    printf("%s : %s\n", name->string, name->valuestring);
    data = cJSON_GetObjectItem(modbus, "data");
    printf("%s\n", data->string);
    size = cJSON_GetArraySize(data);
    for (int i = 0; i < size; i++) // 遍历数组项
    {
        printf("-------------\n");
        cJSON *tmp = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(tmp, "key");
        cJSON *name = cJSON_GetObjectItem(tmp, "name");
        cJSON *addr = cJSON_GetObjectItem(tmp, "addr");
        cJSON *type = cJSON_GetObjectItem(tmp, "type");
        node_arr[i + index].key = key->valueint;
        node_arr[i + index].type = type->valueint;
        node_arr[i + index].dev_type = 1; // 1:modbus
        // node_arr[i + index].old_val.i_val = 20;
        // node_arr[i + index].new_val.i_val = 25;
        node_arr[i + index].ret = -1;
    }
    *total = size + index; // 设备数量
    //***************************
    // STM32
    node_arr[0].new_val.f_val = 23; // 温度
    node_arr[1].new_val.f_val = 40; // 湿度
    node_arr[2].new_val.f_val = 30; // 电量
    node_arr[3].new_val.i_val = 1;  //灯
    // modbus
    node_arr[4].new_val.f_val = 33; // 甲醛
    node_arr[5].new_val.f_val = 34; // 有害气体
    node_arr[6].new_val.i_val = 0;  // 排风开关
    node_arr[7].new_val.i_val = 1;  // 空气净化器开关
    //***************************

    /*======================TODO 数据上报线程======================*/
    pthread_t up_tid;
    if (pthread_create(&up_tid, NULL, upHandler, NULL) != 0) // 创建成功返回0
    {
        perror("upHandler pthread_create err.");
        return -1;
    }
    /*======================TODO 数据存储线程======================*/
    pthread_t store_tid;
    if (pthread_create(&store_tid, NULL, storeHandler, NULL) != 0) // 创建成功返回0
    {
        perror("storeHandler pthread_create err.");
        return -1;
    }
    /*======================TODO 指令处理线程======================*/
    pthread_t instruct_tid;
    if (pthread_create(&instruct_tid, NULL, instructHandler, NULL) != 0) // 创建成功返回0
    {
        perror("upHandler pthread_create err.");
        return -1;
    }
    /*===========================================================*/

    while (1) // 循环检测终端输入，为 quit 退出整个程序
    {
        fgets(msg, sizeof(msg), stdin);
        if (msg[strlen(msg) - 1] == '\n')
        {
            msg[strlen(msg) - 1] = '\0';
        }
        if (strncmp("quit", msg, 4) == 0)
        {
            break;
        }
        // pubmsg.payload = msg;
        // pubmsg.payloadlen = (int)strlen(msg);
        // pubmsg.qos = QOS;
        // pubmsg.retained = 0;
        // MQTTClient_publishMessage(client, TOPIC2, &pubmsg, &token);
    }
    MQTTClient_unsubscribe(client, TOPIC1);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    cJSON_Delete(root); //cJSON_Parse在解析的过程中，会分配很多内存，所以最后要释放
    return 0;
}

// 数据上报线程函数（每次运行只会选择一种上报模式）
void *upHandler(void *arg)
{
    if (type == 0) // 刷新上报
    {
        flashUp();
    }
    else if (type == 1) // 变化上报
    {
        changeUp();
    }
    else if (type == 2) //  周期上报
    {
        circleUp();
    }
}

void flashUp() // 刷新上报（全部上报）
{
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *arr = cJSON_CreateArray();
    cJSON *root2;

    char string[16] = {0};
    for (int i = 0; i < (*total); i++)
    {
        root2 = cJSON_CreateObject();
        cJSON_AddItemToObject(root2, "key", cJSON_CreateNumber(node_arr[i].key));
        if (node_arr[i].type == 1)
        {
            sprintf(string, "%d", node_arr[i].new_val.b_val);
            cJSON_AddItemToObject(root2, "val", cJSON_CreateString(string));
            node_arr[i].old_val.b_val = node_arr[i].new_val.b_val;
        }
        else if (node_arr[i].type == 2)
        {
            sprintf(string, "%d", node_arr[i].new_val.i_val);
            cJSON_AddItemToObject(root2, "val", cJSON_CreateString(string));
            node_arr[i].old_val.i_val = node_arr[i].new_val.i_val;
        }
        else if (node_arr[i].type == 3)
        {
            sprintf(string, "%f", node_arr[i].new_val.f_val);
            cJSON_AddItemToObject(root2, "val", cJSON_CreateString(string));
            node_arr[i].old_val.f_val = node_arr[i].new_val.f_val;
        }
        cJSON_AddItemToArray(arr, root2);
    }
    cJSON_AddItemToObject(root, "data", arr);
    char *p = cJSON_Print(root);

    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, TOPIC2, &pubmsg, &token); // 发布（上传）
    free(p);
    cJSON_Delete(root);
}
// 变化上报
void changeUp()
{
    flashUp(); // 全部上报一次
    while (1)
    {
        sleep(1);
        for (int i = 0; i < (*total); i++)
        {
            // 比较新旧数据，不相同上报
            // 其中memcmp()不同返回1,相同返回0
            if (memcmp(&node_arr[i].new_val, &node_arr[i].old_val, sizeof(node_arr[i].new_val)))
            {
                flashUp();
            }
        }
    }
}
void circleUp() // 周期（每period秒）上报
{
    while (1)
    {
        sleep(period); // 睡眠period秒（period为全局变量，其值为读取点表中的period）
        flashUp();
    }
}
// 数据存储线程函数
void *storeHandler(void *arg)
{
    sqlite3 *db = NULL;
    //打开或创建数据库
    if (sqlite3_open("./data.db", &db) != 0)
    {
        fprintf(stderr, "sqlite3_open err:%s\n", sqlite3_errmsg(db));
        return NULL;
    }
    //在数据库中创建表
    char *errmsg = NULL;
    char sql[128];                     //存放sql语句
    for (int i = 0; i < (*total); i++) // 每个设备创建一个数据库表，表名为key值
    {
        sprintf(sql, "CREATE TABLE dev%d(time INTEGER PRIMARY KEY,key INTEGER,value REAL);", node_arr[i].key);
        if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != 0)
        {
            fprintf(stderr, "create err:%s\n", errmsg);
        }
    }
    sleep(3);
    while (1)
    {
        sleep(10);
        // 获取当前时间
        time_t t1;
        // printf("time=%ld\n", t1);
        // 遍历共享内存数据，插入到数据库表中
        for (int i = 0; i < (*total); i++)
        {
            time(&t1); // 获取时间戳
            // printf("time=%ld\n", t1);
            if (node_arr[i].type == 1) // 数据类型为 bool
            {
                sprintf(sql, "INSERT INTO dev%d VALUES(%ld,%d,%d);",
                        node_arr[i].key, t1, node_arr[i].key, node_arr[i].new_val.b_val);
            }
            else if (node_arr[i].type == 2) // 数据类型为 int
            {
                sprintf(sql, "INSERT INTO dev%d VALUES(%ld,%d,%d);",
                        node_arr[i].key, t1, node_arr[i].key, node_arr[i].new_val.i_val);
            }
            else if (node_arr[i].type == 3) // 数据类型为 float
            {
                sprintf(sql, "INSERT INTO dev%d VALUES(%ld,%d,%f);",
                        node_arr[i].key, t1, node_arr[i].key, node_arr[i].new_val.f_val);
            }
            // 执行sql语句
            if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != 0)
            {
                fprintf(stderr, "insert err:%s", errmsg);
                printf("执行：%s发生了错误\n",sql);
                //return NULL;
            }
        }
        printf("time=%ld 数据插入数据库成功...\n",t1);
    }
}
// 指令处理线程函数
void *instructHandler(void *arg)
{
    while (1) // 轮寻检测指令类型标志位
    {
        if (insFlag == 1) // 采集指令
        {
            insFlag = 0; // 先恢复原值再执行对应函数，便于下次指令的处理
            flashUp();
        }
        else if (insFlag == 2) // 控制指令
        {
            insFlag = 0;
            ctrlFunc();
        }
        else if (insFlag == 3) // 模式修改
        {
            insFlag = 0;
            modeChange();
        }
        else if (insFlag == 4) // 历史数据查询
        {
            insFlag = 0;
            historysearch();
        }
    }
}
// 控制指令函数
void ctrlFunc()
{
    // 拿到设备的key值ctrlkey，可以用共享内存中结构体的dev_type字段来确定是发给哪个设备，然后通过消息队列发送
    for (int i = 0; i < (*total); i++) // 遍历共享内存结构体数组
    {
        if (node_arr[i].key == ctrlkey)
        {
            if (node_arr[i].dev_type == 0) // 0：stm32端
            {
                send_buf.mtype = 1;
                if (msg_queue_send("stm32topic", &send_buf, sizeof(send_buf), 0) < 0)
                {
                    printf("msg_queue_send error\n");
                    return;
                }
            }
            else if (node_arr[i].dev_type == 1) // 1：modbus端
            {
                send_buf.mtype = 2;
                if (msg_queue_send("modbustopic", &send_buf, sizeof(send_buf), 0) < 0)
                {
                    printf("msg_queue_send error\n");
                    return;
                }
            }
        }
    }
}
// 模式修改函数
void modeChange()
{
    FILE *file = fopen("node.json", "r");
    if (file == NULL)
    {
        perror("fopen err");
        return;
    }
    char buf[4096] = {0};
    size_t len = fread(buf, sizeof(char), 4096, file);
    if (len < 0)
    {
        perror("read err");
        return;
    }
    fclose(file); // 先关闭文件，再以"w"模式打开该文件（会清空内容）
    file = fopen("node.json", "w");
    // 把之前文件中json字符串转成cJSON的结构（反序列化），更改后重新序列化写入文件
    cJSON *noderoot = cJSON_Parse(buf);
    if (NULL == noderoot)
    {
        printf("parse err\n");
        return;
    }
    // 获取设备上报属性
    cJSON *report = cJSON_GetObjectItem(noderoot, "report");
    cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(changetype));
    cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(changeperiod));
    char *p = cJSON_Print(noderoot);
    strcpy(buf, p);
    fputs(buf, file); // 更新点表文件
    printf("[系统通知]：点表已修改！！！\n");
    free(p);
    fclose(file);
    cJSON_Delete(noderoot);sleep(1);
    printf("[系统通知]：上报模式已修改，请重启！！！\n");
    exit(0); // 终止整个进程
}
// 历史数据查询函数
void historysearch()
{
    sqlite3 *db = NULL;
    //打开或创建数据库
    char **resultp = NULL;
    int nrow, ncolumn;
    char *errmsg;
    char sql[128];
    if (sqlite3_open("./data.db", &db) != 0)
    {
        fprintf(stderr, "sqlite3_open err:%s\n", sqlite3_errmsg(db));
        return;
    }
    if (historykey == 3011)
    {
        sprintf(sql, "SELECT * FROM dev3011 WHERE time BETWEEN %ld AND %ld;", startTime, endTime);
        if (sqlite3_get_table(db, sql, &resultp, &nrow, &ncolumn, &errmsg) != 0)
        {
            fprintf(stderr, "select err:%s\n", errmsg);
            return;
        }
        if (nrow == 0) //没有查询到
        {
            printf("没有查询到---------\n");
            return;
        }
        else // 查询到了
        {
            printf("nrow = %d\n", nrow);
            int timeindex = 0, valindex = 2; // 返回的resultp数组下标
            MQTTClient_message pubmsg = MQTTClient_message_initializer;
            MQTTClient_deliveryToken token;
            // 序列化
            cJSON *root1 = cJSON_CreateObject();
            cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(4));
            cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
            cJSON_AddItemToObject(root1, "key", cJSON_CreateNumber(historykey));
            cJSON *arr = cJSON_CreateArray();
            cJSON *root2;
            for (int i = 0; i < nrow; i++) // 遍历查找到的数据条数
            {
                root2 = cJSON_CreateObject();
                timeindex += ncolumn;
                valindex += ncolumn;
                cJSON_AddItemToObject(root2, "time", cJSON_CreateNumber(atoi(resultp[timeindex])));
                cJSON_AddItemToObject(root2, "val", cJSON_CreateString(resultp[valindex]));
                cJSON_AddItemToArray(arr, root2);
            }
            cJSON_AddItemToObject(root1, "data", arr);
            char *p = cJSON_Print(root1);
            pubmsg.payload = p;
            pubmsg.payloadlen = (int)strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPIC2, &pubmsg, &token); // 发布（上传）
            free(p);
            cJSON_Delete(root1);
        }
    }
    else if (historykey == 3012)
    {
        sprintf(sql, "SELECT * FROM dev3012 WHERE time BETWEEN %ld AND %ld;", startTime, endTime);
        if (sqlite3_get_table(db, sql, &resultp, &nrow, &ncolumn, &errmsg) != 0)
        {
            fprintf(stderr, "select err:%s\n", errmsg);
            return;
        }
        if (nrow == 0) //没有查询到
        {
            printf("没有查询到---------\n");
            return;
        }
        else // 查询到了
        {
            int timeindex = 0, valindex = 2; // 返回的resultp数组下标
            MQTTClient_message pubmsg = MQTTClient_message_initializer;
            MQTTClient_deliveryToken token;
            // 序列化
            cJSON *root1 = cJSON_CreateObject();
            cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(4));
            cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
            cJSON_AddItemToObject(root1, "key", cJSON_CreateNumber(historykey));
            cJSON *arr = cJSON_CreateArray();
            cJSON *root2;

            for (int i = 0; i < nrow; i++) // 遍历查找到的数据条数
            {
                root2 = cJSON_CreateObject();

                timeindex += ncolumn;
                valindex += ncolumn;
                cJSON_AddItemToObject(root2, "time", cJSON_CreateNumber(atoi(resultp[timeindex])));
                cJSON_AddItemToObject(root2, "val", cJSON_CreateString(resultp[valindex]));
                cJSON_AddItemToArray(arr, root2);
            }
            cJSON_AddItemToObject(root1, "data", arr);
            char *p = cJSON_Print(root1);
            pubmsg.payload = p;
            pubmsg.payloadlen = (int)strlen(p);
            pubmsg.qos = QOS;
            pubmsg.retained = 0;
            MQTTClient_publishMessage(client, TOPIC2, &pubmsg, &token); // 发布（上传）
            free(p);
            cJSON_Delete(root1);
        }
    }
}
